package cn.edu.flink.scala.tutorial.process

import cn.edu.flink.scala.tutorial.source.{ClickSource, Event}
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object ProcessFunctionTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()

    val stream = env.addSource(new ClickSource).assignAscendingTimestamps(_.timestamp)

    stream.process(new ProcessFunction[Event, String] {
      override def processElement(value: Event, ctx: ProcessFunction[Event, String]#Context, out: Collector[String]): Unit = {
        //        if (!value.user.equals("Mary")) return

        out.collect(value.user + " " + value.url + " " + value.timestamp + " " + getRuntimeContext.getIndexOfThisSubtask
          + " " + ctx.timerService().currentProcessingTime() + " " + ctx.timerService().currentWatermark())
      }
    })
      .print()

    env.execute("ProcessFunctionTest")

  }
}
